package lowlevel;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author:RenPu
 * @Date: 2020/3/27 18:01
 * @Version: 1.0
 * @description:
 */
public class WordCountProcesserTwo implements Processor<Integer,String> {
    //创建上下文对象
    ProcessorContext processorContext=null;

    //创建一个map集合用来封装，暂存处理器切割后的数据的结果

    //map集合储存中间结果状态的缺点 1：map运行在jvm里，程序停止或者重启，数据会有丢失。2：map的储存容量有限为 1<<30 相当于int的最大取值范围
    Map<String,Long> map=null;

    /**
     * 初始化方法，联系上下文环境
     * @param processorContext
     */
    @Override
    public void init(ProcessorContext processorContext) {

        //初始化上下文环境对象
        this.processorContext=processorContext;

        //初始化map集合
        this.map=new HashMap<>(20);

        //调度任务，每个1秒触发一次，向下游发送当前完成的处理结果
        this.processorContext.schedule(Duration.ofSeconds(1), PunctuationType.STREAM_TIME, new Punctuator() {
            @Override
            public void punctuate(long l) {
                map.forEach((k,v)->{
                    //转发
                    processorContext.forward(k,v);
                });

            }
        });



    }


    /**
     * 单词切割的方法
     * @param key  一条流数据的key
     * @param value value值
     *
     */
    @Override
    public void process(Integer key, String value) {

        String[] words = value.split(" ");

        for (String word : words) {

            Long num = map.getOrDefault(word, 0L);
            num++;
        }
        processorContext.commit();

    }


    /**
     * 释放资源
     */
    @Override
    public void close() {

    }
}
